﻿using log4net.Appender;
using log4net.Core;
using NsqSharp;
using NsqSharp.Api;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Runtime.Serialization.Formatters.Binary;
using System.Threading;
using System.Threading.Tasks;

namespace log4net.extension.nsq
{
    public class NSQAppender : AppenderSkeleton
    {
        protected ConcurrentDictionary<string, Producer> m_producerContainer = null;
        protected ConcurrentDictionary<string, NsqdHttpClient> m_nsqdClientContainer = null;
        protected ConcurrentDictionary<string, Producer> m_errorContainer = null;
        protected string m_localIp = null;
        protected static int increNum = 0;
        public override void ActivateOptions()
        {
            base.ActivateOptions();

            m_producerContainer = new ConcurrentDictionary<string, Producer>();
            m_nsqdClientContainer = new ConcurrentDictionary<string, NsqdHttpClient>();
            m_errorContainer = new ConcurrentDictionary<string, Producer>();

            var config = new NsqSharp.Config();
            config.DialTimeout = TimeSpan.FromMilliseconds(200);
            if (NsqLookupAddresses != null)
            {
                var array = NsqLookupAddresses.Split(',');
                for (int i = 0; i < array.Length; i++)
                {
                    try
                    {
                        NsqLookupdHttpClient client = new NsqLookupdHttpClient(array[i], TimeSpan.FromSeconds(1));
                        var nodes = client.GetNodes();
                        foreach (var n in nodes)
                        {
                            var tcpAddress = string.Format("{0}:{1}", n.BroadcastAddress, n.TcpPort);
                            var httpAddress = string.Format("{0}:{1}", n.BroadcastAddress, n.HttpPort);

                            var key = tcpAddress;

                            m_producerContainer.TryAdd(key, new Producer(tcpAddress, config));
                            m_nsqdClientContainer.TryAdd(key, new NsqdHttpClient(httpAddress, TimeSpan.FromSeconds(1)));
                        }
                        break;
                    }
                    catch (Exception ex)
                    {
                        ErrorHandler.Error("Exception while get nodes from nsqlookup address", ex);
                    }
                }
            }
            m_localIp = GetLocalIp();
        }
        protected override void Append(LoggingEvent loggingEvent)
        {
            var kv = SelectProducer();
            Producer producer = kv.Value;
            try
            {
                if (producer != null)
                {
                    var message = GetLogEntity(loggingEvent);
                    producer.Publish(this.TopicName, Serialize(message));
                }
            }
            catch (Exception ex)
            {
                HealthCheck(kv);
                ErrorHandler.Error("Exception while writing to nsq", ex);
            }
        }

        protected virtual KeyValuePair<string, Producer> SelectProducer()
        {
            Interlocked.Increment(ref increNum);
            var index = increNum % m_producerContainer.Count;
            return m_producerContainer.ElementAtOrDefault(Math.Abs(index));
        }
        protected virtual Log4netLogEntity GetLogEntity(LoggingEvent loggingEvent)
        {
            string message = string.Empty;
            string exceptionStr = loggingEvent.GetExceptionString();
            if (exceptionStr != null && exceptionStr.Length > 0)
                message = string.Format("{0}\r\n{1}", loggingEvent.RenderedMessage, exceptionStr);
            else
                message = loggingEvent.RenderedMessage;

            return new Log4netLogEntity
            {
                LogTime = loggingEvent.TimeStamp.ToString("yyyy-MM-dd HH:mm:ss"),
                Level = loggingEvent.Level.DisplayName,
                ServerIp = m_localIp,
                Message = message
            };
        }
        protected virtual string GetLocalIp()
        {
            var ipArray = Dns.GetHostEntry(Dns.GetHostName()).AddressList;
            if (ipArray != null && ipArray.Length > 0)
            {
                var ip4 = ipArray.FirstOrDefault(p => p.AddressFamily == System.Net.Sockets.AddressFamily.InterNetwork);
                if (ip4 != null)
                {
                    return ip4.ToString();
                }
            }
            return string.Empty;
        }
        protected virtual byte[] Serialize(Log4netLogEntity data)
        {
            if (data == null)
                return null;

            using (MemoryStream ms = new MemoryStream())
            {
                BinaryFormatter formatter = new BinaryFormatter();
                formatter.Serialize(ms, data);
                ms.Position = 0;
                byte[] bytes = new byte[ms.Length];
                ms.Read(bytes, 0, bytes.Length);
                return bytes;
            }
        }

        private static object _obj = new object();
        private static bool isHealthChecking = false;
        protected virtual void HealthCheck(KeyValuePair<string, Producer> badProducerKV)
        {
            m_producerContainer.TryRemove(badProducerKV.Key, out Producer p);
            m_errorContainer.TryAdd(badProducerKV.Key, badProducerKV.Value);

            if (isHealthChecking) return;
            lock (_obj)
            {
                if (isHealthChecking) return;
                Task.Run(() =>
                {
                    isHealthChecking = true;

                    while (true)
                    {
                        try
                        {
                            List<KeyValuePair<string, Producer>> okProducerList = new List<KeyValuePair<string, Producer>>();
                            //re check error producer
                            foreach (var kv in m_errorContainer)
                            {
                                var client = m_nsqdClientContainer[kv.Key];
                                try
                                {
                                    var result = client.Ping();
                                    if (string.Equals(result, "ok", StringComparison.CurrentCultureIgnoreCase))
                                    {
                                        okProducerList.Add(new KeyValuePair<string, Producer>(kv.Key, kv.Value));
                                        ErrorHandler.Error(string.Format("error node {0} reconnect success", kv.Key));
                                    }
                                }
                                catch
                                { }
                            }

                            foreach (var kv in okProducerList)
                            {
                                m_errorContainer.TryRemove(kv.Key, out Producer p1);
                                m_producerContainer.TryAdd(kv.Key, kv.Value);
                            }

                            if (m_errorContainer.Count == 0) break;
                        }
                        catch { }
                        Thread.Sleep(10000);
                    }

                    isHealthChecking = false;
                });
            }
        }

        protected override void OnClose()
        {
            base.OnClose();
            if (m_producerContainer != null)
            {
                foreach (var producer in m_producerContainer.Values)
                {
                    producer.Stop();
                }
            }
        }

        #region property

        public string TopicName { get; set; }
        public string NsqLookupAddresses { get; set; }

        #endregion
    }
}
